package com.gitee.flwwsg.rpc.transport.netty;

import com.gitee.flwwsg.rpc.transport.InFlightRequests;
import com.gitee.flwwsg.rpc.transport.ResponseFuture;
import com.gitee.flwwsg.rpc.transport.command.Command;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ResponseInvocation extends SimpleChannelInboundHandler<Command> {

    private static final Logger logger = LoggerFactory.getLogger(ResponseInvocation.class);
    private final InFlightRequests inFlightRequests;

    ResponseInvocation(InFlightRequests inFlightRequests) {
        this.inFlightRequests = inFlightRequests;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Command msg) throws Exception {
        //移除已经处理的消息
        ResponseFuture future = inFlightRequests.remove(msg.getHeader().getRequestId());
        if (null != future) {
            future.getFuture().complete(msg);
        } else {
            logger.warn("Drop response: {}", msg);
            throw new Exception("time out for response");
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.warn("Exception: ", cause);
        // 谁引发异常，谁处理
        // super.exceptionCaught(ctx, cause);
        Channel channel = ctx.channel();
        if (channel.isActive()) {
            ctx.close();
        }
    }
}
